import java.net.URI
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
 * 有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死
 * 优雅的关闭：先切断读的数据，再把计算结束后关闭。
 */
object Stop {

  def main(args: Array[String]): Unit = {


    val conf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 设置优雅的关闭
    conf.set("spark.streaming.stopGracefullyOnShutdown", "true")


    ssc.socketTextStream("usdp-o3tbdsfp-monitor1", 4444)
      .flatMap(_.split(" "))
      .map((_, 1))
      .print()

    // 开启监控程序
    new Thread(new MonitorStop(ssc)).start()


    ssc.start()
    ssc.awaitTermination()
  }
}

//监控程序
class MonitorStop(ssc: StreamingContext) extends Runnable {

  override def run(): Unit = {

    //获取HDFS文件系统
    val fs: FileSystem = FileSystem.get(new URI("hdfs://usdp-o3tbdsfp-master2:8020"), new Configuration(), "hadoop")

    while (true) {
      Thread.sleep(5000)

      val result: Boolean = fs.exists(new Path("hdfs://usdp-o3tbdsfp-master2:8020/stopSpark"))

      if (result) {

        val state: StreamingContextState = ssc.getState()
        // 获取当前任务是否正在运行
        if (state == StreamingContextState.ACTIVE) {
          // 优雅关闭
          ssc.stop(stopSparkContext = true, stopGracefully = true)
          System.exit(0)
        }
      }
    }
  }
}
